package com.example.study;

import androidx.appcompat.app.AppCompatActivity;

import android.os.Bundle;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;

import io.reactivex.rxjava3.annotations.NonNull;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableEmitter;
import io.reactivex.rxjava3.core.ObservableOnSubscribe;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Consumer;
import io.reactivex.rxjava3.functions.Function;
import io.reactivex.rxjava3.plugins.RxJavaPlugins;

public class MainActivity extends AppCompatActivity {

    @Override
    protected void onCreate(Bundle savedInstanceState) {
        super.onCreate(savedInstanceState);
        setContentView(R.layout.activity_main);
        Thread thread = new Thread();
        thread.setPriority(Thread.NORM_PRIORITY);
        thread.start();

        /* 不知道的还以为你是哪一位大神 */
        ArrayBlockingQueue arrayBlockingQueue;
        LinkedBlockingQueue blockingQueue;
        PriorityBlockingQueue blockingQueue1;
        DelayQueue delayQueue;
        SynchronousQueue synchronousQueue;
        LinkedBlockingQueue linkedBlockingQueue;

        RxJavaPlugins.setOnObservableAssembly(new Function<Observable, Observable>() {
            @Override
            public Observable apply(Observable observable) throws Throwable {
                //整个项目 全局 监听 到底有多少地方使用Rxjava+observable
                return observable;
            }
        });

        Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> emitter) throws Throwable {
                emitter.onNext("String");//发射器，分发给下游的信息
            }
        }).doOnNext(new Consumer<String>() {
            @Override
            public void accept(String s) throws Throwable {
                //中间插入做其他事情
            }
        })
                .map(new Function<String, String>() {
                    @Override
                    public String apply(String s) throws Throwable {
                        return "String Map";
                    }
                }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {

            }

            @Override
            public void onNext(@NonNull String s) {

            }

            @Override
            public void onError(@NonNull Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    private class Syn extends AbstractQueuedSynchronizer {
        @Override
        protected boolean tryAcquire(int arg) {
            compareAndSetState(0, 1);
            return super.tryAcquire(arg);
        }
    }
}